package com.codejiwei.flink.zk;

import org.apache.flink.shaded.curator4.org.apache.curator.framework.CuratorFramework;
import org.apache.flink.shaded.curator4.org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.flink.shaded.curator4.org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.flink.shaded.curator4.org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.flink.shaded.curator4.org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.IOException;

/**
 * author: codejiwei
 * date: 2023/9/7
 * desc:
 **/
public class ZookeeperSource implements SourceFunction<byte[]> {

    private final String connect;
    private final String path;
    private transient NodeCache nodeCache;

    public ZookeeperSource(String connect, String path) {
        this.connect = connect;
        this.path = path;
        CuratorFramework curator = CuratorFrameworkFactory.newClient(connect, new ExponentialBackoffRetry(1000, 3));
        nodeCache = new NodeCache(curator, path);
    }

    @Override
    public void run(SourceContext<byte[]> ctx) throws Exception {
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                byte[] data = nodeCache.getCurrentData().getData();
                ctx.collect(data);
            }
        });
        nodeCache.start();
        synchronized (this) {
            //阻塞等待,直到被取消
            wait();
        }

    }

    @Override
    public void cancel() {

        try {
            // 关闭NodeCache对象
            nodeCache.close();
        } catch (IOException exception) {
            throw new RuntimeException("关闭NodeCache失败!", exception);
        }
        synchronized (this) {
            // 唤醒等待的线程
            notifyAll();
        }
    }
}
